use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

use crate as honeybee;
use crate::etcd;
use crate::prelude::message;
use crate::prelude::*;
use crate::Result;

use super::etcd_usc_server::UscServerInfo;

pub struct EtcdUscClientActor {
    addr: Option<Addr<Self>>,
    etcd_client: etcd_client::Client,
    node_type: String,
    node_name: String,
    message_register: Arc<MessageRegister>,
    ready_usc_endpoints: BTreeMap<String, String>,
}

impl EtcdUscClientActor {
    pub fn new(node_type: &str, node_name: &str, message_register: Arc<MessageRegister>) -> Self {
        Self {
            addr: None,
            etcd_client: etcd::get_client(),
            node_type: node_type.into(),
            node_name: node_name.into(),
            message_register,
            ready_usc_endpoints: BTreeMap::default(),
        }
    }

    async fn try_ready(&mut self, info: UscServerInfo) -> Result<bool> {
        self.ready_usc_endpoints
            .insert(info.node_name, info.node_endpoint);
        if self.ready_usc_endpoints.len() == info.node_total_number {
            let endpoints: Vec<String> = self
                .ready_usc_endpoints
                .iter()
                .map(|(_, endpoint)| endpoint.clone())
                .collect();
            let _ = TcpConnectersActor::new(
                "usc_server",
                &self.node_type,
                &self.node_name,
                &endpoints,
                Duration::from_secs(3),
                self.message_register.clone(),
            )
            .start()
            .await?;

            if let Some(addr) = self.addr.take() {
                let _ = addr.stop(None);
            }
            return Ok(true);
        }
        Ok(false)
    }
}

#[async_trait::async_trait]
impl Actor for EtcdUscClientActor {
    fn name(&self) -> &str {
        "EtcdUscClientActor"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        info!("etcd usc server actor:{} started", self.name());
        let addr = ctx.address();
        addr.send(Init)?;
        self.addr = Some(addr);
        Ok(())
    }

    async fn stopped(&mut self, _ctx: &mut Context<Self>) {
        info!("{} stoped", self.name());
    }
}

#[message]
pub struct Init;
#[async_trait::async_trait]
impl Handler<Init> for EtcdUscClientActor {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: Init) -> Result<()> {
        let get_opt = etcd_client::GetOptions::new().with_prefix();
        let resp = self.etcd_client.get("/usc", Some(get_opt)).await?;
        for kv in resp.kvs() {
            let key = kv.key_str()?;
            let value = kv.value_str()?;
            let info: UscServerInfo = serde_json::from_str(value)?;
            info!("etcd usc client init get key:{} value:{:?}", key, info);
            if self.try_ready(info).await? {
                return Ok(());
            }
        }
        let watch_opt = etcd_client::WatchOptions::new().with_prefix();
        let (_, stream) = self.etcd_client.watch("/usc", Some(watch_opt)).await?;
        ctx.add_stream(stream);
        Ok(())
    }
}

#[async_trait::async_trait]
impl StreamHandler<std::result::Result<etcd_client::WatchResponse, etcd_client::Error>>
    for EtcdUscClientActor
{
    async fn handle(
        &mut self,
        ctx: &mut Context<Self>,
        response: std::result::Result<etcd_client::WatchResponse, etcd_client::Error>,
    ) {
        let resp = match response {
            Ok(resp) => resp,
            Err(err) => {
                error!("{} err:{}", self.name(), err);
                return;
            }
        };

        if resp.canceled() {
            warn!("{} canceled", self.name());
            ctx.abort_streams();
            return;
        }

        for event in resp.events() {
            info!("{} event type: {:?}", self.name(), event.event_type());
            if event.event_type() == etcd_client::EventType::Put {
                if let Some(kv) = event.kv() {
                    if let (Ok(key), Ok(value)) = (kv.key_str(), kv.value_str()) {
                        if let Ok(info) = serde_json::from_str::<UscServerInfo>(value) {
                            info!("etcd usc client init get key:{} value:{:?}", key, info);
                            if let Err(err) = self.try_ready(info).await {
                                error!("{} err:{}", self.name(), err);
                            }
                        }
                    }
                }
            }
        }
    }

    async fn finished(&mut self, _ctx: &mut Context<Self>) {
        info!("{} stoped", self.name());
    }
}
